//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#include <sys/stat.h>
#include <sys/times.h>
#include <sys/wait.h>
#include <sys/types.h>

#include <unistd.h>
#include <fcntl.h>

#include <time.h>
#include <errno.h>

#include "rutines.h"
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
#define SIZE                   1000
#define MAX_CONCURRENT_PROCESS 4
#define MAX_TICKECT_READ       100
#define MODE_NORMAL            0
#define MODE_FAST              1
#define DELAY                  2000000L
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
void signalUSR2 (int sig);
void orderClusters (int pid);
void delay (void);
void addBuffer(char** buffer, char* content, int size);
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
int         numberProcess = 0;
int         forkProcess   = 0;
int         pidsArray [MAX_CONCURRENT_PROCESS];
int         pipes     [MAX_CONCURRENT_PROCESS][2];
int         inputFile;
int         status        = 1;
int         mode          = MODE_NORMAL;
int         blockSize     = 0;
const char* blockArgv     = 0;
const char* namefileOut   = 0;
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
int main (int argc, char** argv) {
  
  int       pidFork;
  int       i;
  int       sig;
  int       noend;
  int       x;
  int       counter;
  int       file;
  int       fileOut;

  char      buffer [MAXSTR];
    
  sigset_t  signalVar;

  if (argc < 4) 
    error("Missing Arguments (./clusterHugeDB <input smi file> <out file - ID cluster assigment> <block size>");

  signal(SIGUSR2, signalUSR2);
  
  for (i = 0; i < argc; i++) 
    if (!memcmp(argv[i], "fast", strlen("fast"))) 
      mode = MODE_FAST;
  
  blockSize   = atoi(argv[3]);
  blockArgv   = argv[3];
  namefileOut = argv[2];
  
  fileOut = open (namefileOut, O_CREAT | O_WRONLY | O_TRUNC, 0644);
  close(fileOut);

  for (i = 0; i < MAX_CONCURRENT_PROCESS; i++) {
    
    if (pipe(pipes[i]) < 0)
      error("Error in pipe");
    
    switch (pidFork = fork()) {
      
      case -1 : error("Error in fork");
      
      case  0 : dup2(pipes[i][0], STDIN_FILENO);
                sprintf(buffer, "%d.dat", getpid());
                unlink(buffer);
                file = open(buffer, O_CREAT | O_WRONLY | O_TRUNC, 0644);
                dup2(file, STDOUT_FILENO);
                close(file);
                
                close(pipes[i][0]);
                close(pipes[i][1]);
                
                if (mode == MODE_FAST)
                  execl("./computeCluster", "./computeCluster", argv[3], "fork", "fast", (char *)0);
                else
                  execl("./computeCluster", "./computeCluster", argv[3], "fork", (char *)0);
                
                error("Error in execl");
      
      default : close(pipes[i][0]);
      
                pidsArray[forkProcess++] = pidFork; 
                sigemptyset(&signalVar);
                sigaddset(&signalVar, SIGUSR1);
                sigprocmask(SIG_BLOCK, &signalVar, NULL);
                sigwait(&signalVar, &sig);
                sigprocmask(SIG_UNBLOCK, &signalVar, NULL);
                
                break;
                
    }
    
  }
  
  sigemptyset(&signalVar); 
  sigaddset(&signalVar, SIGCHLD);
  sigprocmask(SIG_BLOCK, &signalVar, NULL);
  
  if ((inputFile = open(argv[1], O_RDONLY)) < 0)
    error("Error in open");
  
  for (i = 0, counter = 0, noend = 1; i < MAX_CONCURRENT_PROCESS; i++) {
    
    if (!noend) { kill(pidsArray[i], SIGKILL); continue; }
    
    do {
      
      if (counter++ >= blockSize) { counter = 0; break; }
    
      x = 0;
      
      while ((noend = read(inputFile, &buffer[x], 1)) == 1 && buffer[x++] != '\n');
      if (!noend) { status = 0; break; }
      
      buffer[x] = '\0';
      
      if (write(pipes[i][1], buffer, strlen(buffer)) < 0)
        error("Error in write");
      
    } while (noend);
    
    close(pipes[i][1]);
    
  }
  
  sigprocmask(SIG_UNBLOCK, &signalVar, NULL);

  if (forkProcess) 
    while (1) {
      if (!forkProcess) break;
      pause();
    }
  
  close(inputFile);
  return EXIT_SUCCESS;
  
}
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
void orderClusters (int pid) {

  int         pidFork;
  int         status;
  int         p [2];
  int         fileOut;
  int         i;
  int         numberParameters = 3;
  
  char        buffer [MAXSTR];
  char*       list_param [numberParameters];
  
  sigset_t    signalVar;
  
  sigemptyset(&signalVar); 
  sigaddset(&signalVar, SIGUSR2);
  sigprocmask(SIG_BLOCK, &signalVar, NULL);

  sprintf(buffer, "%d.dat\0", pid);
  
  list_param[0] = (char*) malloc((strlen("cat") + 1) * sizeof(char));
  strcpy(list_param[0], "cat");
  list_param[strlen("cat")]  = '\0';
  list_param[1] = (char*) malloc((strlen(buffer) + 1) * sizeof(char));
  strcpy(list_param[1], buffer);
  list_param[strlen(buffer)] = '\0';
  list_param[2] = NULL;
  
  if (pipe(p) < 0)
    error("Error in pipe");
  
  switch (pidFork = fork()) {

    case -1 : error("Error in fork");
    
    case  0 : dup2(p[1], STDOUT_FILENO);
              close(p[0]); 
              close(p[1]);
              execvp(list_param[0], list_param);
              error("Error in execvp (cat)");

    default : break;
    
  }

  switch (pidFork = fork()) {

    case -1 : error("Error in fork");
    
    case  0 : if ((fileOut = open(namefileOut, O_WRONLY | O_APPEND)) < 0) 
                error("Error in open");

              dup2(p[0],    STDIN_FILENO);
              dup2(fileOut, STDOUT_FILENO);
              close(fileOut);
              close(p[0]);
              close(p[1]);
              execlp("sort", "sort", "-n", "-k2", (char *)0);
              error("Error in execlp");

    default : break;
    
  }

  pidFork = wait (&status); 
  
  close (p[0]);
  close (p[1]);

  pidFork = wait (&status); 
  
  if (list_param[0]) free(list_param[0]);
  list_param[0] = (char*) malloc((strlen("rm") + 1) * sizeof(char));
  strcpy(list_param[0], "rm");
  list_param[strlen("rm")] = '\0';

  switch (pidFork = fork()) {

    case -1 : error("Error in fork");
    
    case  0 : execvp(list_param[0], list_param);
              error("Error in execvp");

    default : pidFork = wait (&status);  break;
    
  }
  
  for (i = 0; i < numberParameters; i++)
    if (list_param[i]) 
      free(list_param[i]);
  
  sigprocmask (SIG_UNBLOCK, &signalVar, NULL);
  
}
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
void signalUSR2 (int sig) {
  
  int         pidFork;
  int         st;
  int         i;
  int         position;
  int         file;
  int         counter = 0;
  int         noend;
  int         x;
  
  sigset_t    signalVar;
  sigset_t    signalWait;
  
  char        buffer [MAXSTR];
  char*       cumulBuffer = NULL;
  
  sigemptyset(&signalVar); 
  sigaddset(&signalVar, SIGUSR2);
  sigprocmask(SIG_BLOCK, &signalVar, NULL);
  
  signal(SIGUSR2, signalUSR2);

  forkProcess--;
  
  pidFork = wait(&st);
  
  orderClusters(pidFork);
  
  for (i = 0; i < MAX_CONCURRENT_PROCESS; i++)
    if (pidsArray[i] == pidFork) {
      pidsArray[position = i] = 0;
      break;
    }
  
  do {

    if (counter++ >= blockSize) break;

    x = 0;
    while ((noend = read(inputFile, &buffer[x], 1)) == 1 && buffer[x++] != '\n');
    if (!noend) { status = 0; break; }
    buffer[x] = '\0';

    addBuffer(&cumulBuffer, buffer, strlen(buffer));
    
  } while (noend);
  
  if (status && cumulBuffer) {
   
    if (pipe(pipes[position]) < 0)
      error("Error in pipe");
    
    switch (pidFork = fork()) {
      
      case -1 : error("Error in fork");
      
      case  0 : dup2(pipes[position][0], STDIN_FILENO);
      
                sprintf(buffer, "%d.dat", getpid());
                unlink(buffer);
                file = open(buffer, O_CREAT | O_WRONLY | O_TRUNC, 0644);
                dup2(file, STDOUT_FILENO);
                close(file);
                
                close(pipes[position][0]);
                close(pipes[position][1]);
                
                if (mode == MODE_FAST)
                  execl("./computeCluster", "./computeCluster", blockArgv, "fork", "fast", (char *)0);
                else
                  execl("./computeCluster", "./computeCluster", blockArgv, "fork", (char *)0);
                
                error("Error in execl");
      
      default : close(pipes[position][0]);
      
                pidsArray[position] = pidFork; 
                forkProcess++;
                
                sigemptyset(&signalWait);
                sigaddset(&signalWait, SIGUSR1);
                sigprocmask(SIG_BLOCK, &signalWait, NULL);
                sigwait(&signalWait, &sig);
                sigprocmask(SIG_UNBLOCK, &signalWait, NULL);
                
                break;
                
    }
    
    if (write(pipes[position][1], cumulBuffer, strlen(cumulBuffer)) < 0)
      error("Error in write");
    
    close(pipes[position][1]);

  }  

  sigprocmask (SIG_UNBLOCK, &signalVar, NULL);
  
}
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
void delay (void) {
  
  struct timespec t;

  t.tv_sec  = 0;
  t.tv_nsec = DELAY;
  
  nanosleep(&t, NULL);

}
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
void addBuffer(char** buffer, char* content, int size) {
  
  char* buffer1      = NULL;
  int   sizeAllocate = 0;
  
  if (*buffer) {
    sizeAllocate = size + strlen(*buffer) + 1;
    buffer1      = (char*) malloc(sizeAllocate * sizeof(char));
    if (!buffer1) error("Error in malloc");
    sprintf(buffer1, "%s%s\0", *buffer, content);
    free(*buffer);
    *buffer      = buffer1;
  }
  else {
    sizeAllocate = size + 1;
    *buffer      = (char*) malloc(sizeAllocate * sizeof(char));
    if (!(*buffer)) error("Error in malloc");
    strcpy(*buffer, content);
    (*buffer)[sizeAllocate - 1] = '\0';
  }

}
//---------------------------------------------------------------------------------------------------------------------------------------------------------//
//---------------------------------------------------------------------------------------------------------------------------------------------------------//

